# +----------------------------------------------------------------------
# | ChatWork智能聊天办公系统
# +----------------------------------------------------------------------
# | 软件声明: 本系统并非自由软件,未经授权任何形式的商业使用均属非法。
# | 版权保护: 任何企业和个人不允许对程序代码以任何形式任何目的复制/分发。
# | 授权要求: 如有商业使用需求,请务必先与版权所有者取得联系并获得正式授权。
# +----------------------------------------------------------------------
# | Author: ChatWork Team <2474369941@qq.com>
# +----------------------------------------------------------------------
import base64
import hashlib
import json
import os
import time
import uuid
import string
import random
import mimetypes
import aiofiles
import aiohttp
from PIL import Image
from datetime import datetime
from typing import Optional, Tuple


class ToolsUtil:
    """ 通用工具 """

    @staticmethod
    def make_uuid() -> str:
        """
        生成UUID的十六进制字符串表示。

        Returns:
            str: UUID的十六进制字符串。

        Author:
            zero
        """
        return uuid.uuid4().hex

    @staticmethod
    def make_token(salt: str = "") -> str:
        """
        生成一个唯一的Token字符串。

        Args:
            salt (str): 加密的盐。

        Returns:
            str: 唯一的Token字符串。

        Author:
            zero
        """
        ms = int(time.time() * 1000)
        token = ToolsUtil.make_md5_str(f"{ToolsUtil.make_uuid()}{ms}{ToolsUtil.make_rand_char(8)}")
        token_secret = f"{token}{ToolsUtil.make_rand_char(20)}{salt}"
        return f"{ToolsUtil.make_md5_str(token_secret)}{ToolsUtil.make_rand_char(6)}"

    @staticmethod
    def make_md5_str(val: str, salt: str = "") -> str:
        """
        生成一个给定值和盐的MD5加密字符串。

        Args:
            val (str): 需要加密的字符串。
            salt (str): 用于增加加密复杂度的盐值。默认为空字符串。

        Returns:
            str: 加密后的MD5字符串。

        Author:
            zero
        """
        m = hashlib.md5()
        m.update((val + salt).encode("utf-8"))
        return m.hexdigest()

    @classmethod
    def make_md5_pwd(cls, pwd: str, salt: Optional[str] = None) -> Tuple[str, str] or str:
        """
        生成加密后的密码。

        Args:
            pwd (str): 需要加密的密码。
            salt (str): 用于增加加密复杂度的盐值。

        Returns:
            str or tuple: 加密后的MD5字符串 和 盐。

        Author:
            zero
        """
        _salt: str = ToolsUtil.make_rand_char(6) if salt is None else salt
        password: str = ToolsUtil.make_md5_str(pwd, _salt)
        if salt is None:
            return _salt, password
        return password

    @staticmethod
    def make_rand_char(length: int) -> str:
        """
        生成指定长度的随机字母和数字字符串。

        Args:
            length (int): 字符串的长度。

        Returns:
            str: 随机生成的字符串。

        Author:
            zero
        """
        str_list = random.sample(string.digits + string.ascii_letters, length)
        random_str = "".join(str_list)
        return random_str

    @staticmethod
    def make_rand_digit(length: int) -> str:
        """
        生成指定长度的随机数字字符串。

        Args:
            length (int): 字符串的长度。

        Returns:
            str: 随机生成的数字字符串。

        Author:
            zero
        """
        random_digits = "".join(random.choices(string.digits, k=length))
        return random_digits

    @staticmethod
    async def make_rand_sn(model, field: str, length: int = 10, scene: str = "digits") -> str:
        """
        生成不重复的随机字符串。

        Args:
            model (any): 模型实体。
            field (str): 字段名。
            length (int): 长度。
            scene (str): 场景: [digits, string, char]。

        Returns:
            str: 随机生成的不重复字符串。

        Author:
            zero
        """
        if scene == "digits":    # 纯数字(0~9)
            random_digits = "".join(random.choices(string.digits, k=length))
        elif scene == "string":  # 纯英文字母
            random_digits = "".join(random.choices(string.ascii_letters, k=length))
        elif scene == "char":    # 数字+字母
            str_list = random.sample(string.digits + string.ascii_letters, length)
            random_digits = "".join(str_list)
        else:
            raise Exception("Unsupported random type ToolUtil.make_rend_sn()")

        if model and await model.exists(**{field: random_digits}):
            return await ToolsUtil.make_rand_sn(model, field, length, scene)
        return random_digits

    @staticmethod
    async def make_order_sn(model, field: str, prefix: str = "", rand_suffix_length: int = 6, pool=None) -> str:
        """
        生成不重复的随机订单号。

        Args:
            model (any): 模型实体。
            field (str): 字段名。
            prefix (str): 前缀名称。
            rand_suffix_length (int): 后缀长度。
            pool (str): 随机池。

        Returns:
            str: 随机生成的不重复订单号。

        Author:
            zero
        """
        if pool is None:
            pool = string.digits

        suffix = "".join(random.choice(pool) for _ in range(rand_suffix_length))
        sn = f"{prefix}{datetime.now().strftime('%Y%m%d%H%M%S')}{suffix}"

        if await model.exists(**{field: sn}):
            return await ToolsUtil.make_order_sn(model, field, prefix, rand_suffix_length, pool)

        return sn

    @classmethod
    async def write_file(cls, path: str,
                         content: any,
                         make_dirs: bool = True,
                         mode: any = "w",
                         encoding: Optional[str] = "UTF-8"):
        """
        异步写入文件。

        Args:
            path (str): 完整文件路径。
            content (str|bytes): 要写入的内容 。
            make_dirs (bool): 自动创建不存在目录。
            mode (str): 写入模式 (wb=字节写入, w=文本写入)。
            encoding (str): 字符编码 (UTF-8)。

        Author:
            zero
        """
        if make_dirs:
            directory_path = os.path.dirname(path)
            if not os.path.exists(directory_path):
                os.makedirs(directory_path)
        async with aiofiles.open(path, mode=mode, encoding=encoding) as file:
            await file.write(content)

    @classmethod
    async def read_file(cls, path: str):
        """
        异步读文件内容。

        Args:
            path (str): 完整文件路径。

        Returns:
            str: 文本内容

        Author:
            zero
        """
        async with aiofiles.open(path, mode="r", encoding="utf-8") as f:
            content = await f.read()
            return content

    @classmethod
    async def get_curl(cls, url: str, headers: dict = None, data: dict = None, timeout=60, ssl=False):
        """
        发起GET请求

        Args:
            url (str): 网络文件地址
            headers (dict): 请求头参数
            data (dict): Body参数
            timeout (int): 超时时间(s)
            ssl (bool): ssl

        Author:
            zero
        """
        async with aiohttp.ClientSession(
            connector=aiohttp.TCPConnector(ssl=ssl),
            timeout=aiohttp.ClientTimeout(total=timeout),
            headers=headers
        ) as session:
            try:
                async with session.get(url=url, params=data) as response:
                    results = await response.text()
                    if response.status == 404:
                        raise Exception("API 404 Not Found")
                    return json.loads(results)
            except aiohttp.client_exceptions.ClientResponseError as err:
                raise Exception(err.status, err.message)

    @classmethod
    async def post_curl(cls, url: str, headers: dict = None, data: dict = None, timeout=60, ssl=False):
        """
        发起POST请求

        Args:
            url (str): 网络文件地址
            headers (dict): 请求头参数
            data (dict): Body参数
            timeout (int): 超时时间(s)
            ssl (bool): ssl

        Author:
            zero
        """
        async with aiohttp.ClientSession(
            connector=aiohttp.TCPConnector(ssl=ssl),
            timeout=aiohttp.ClientTimeout(total=timeout),
            headers=headers
        ) as session:
            try:
                async with session.post(url=url, json=data) as response:
                    results = await response.text()
                    if response.status == 404:
                        raise Exception("API 404 Not Found")
                    return json.loads(results)
            except aiohttp.client_exceptions.ClientResponseError as err:
                raise Exception(err.status, err.message)

    @classmethod
    async def download_file(cls, url: str, save_path: str):
        """
        下载网络文件

        Args:
            url (str): 网络文件地址
            save_path (str): 保存路径(如: /www/images/test.png)

        Author:
            zero
        """
        async with aiohttp.ClientSession() as session:
            async with session.get(url) as response:
                if response.status == 200:
                    # 确保保存目录存在
                    os.makedirs(os.path.dirname(save_path), exist_ok=True)
                    # 写入模式保存文件
                    async with aiofiles.open(save_path, "wb") as f:
                        while True:
                            chunk = await response.content.read(2048)
                            if not chunk:
                                break
                            await f.write(chunk)
                    return save_path
                else:
                    raise Exception(response.status)

    @classmethod
    async def convert_to_base64(cls, path: str):
        """
        把文件转成Base64

        Args:
            path (str): URL/PATH。

        Returns:
            str|int: base64内容,失败返回404

        Author:
            zero
        """
        if path.startswith("/"):
            if not os.path.exists(path):
                return 404
            async with aiofiles.open(path, mode="rb") as f:
                chunks = []
                while True:
                    chunk = await f.read(2048)
                    if not chunk:
                        break
                    chunks.append(chunk)
                file_content = b''.join(chunks)
            mime_type, _ = mimetypes.guess_type(str(path))
            base64_encoded = base64.b64encode(file_content).decode('utf-8')
            return f"data:{mime_type};base64,{base64_encoded}"
        else:
            async with aiohttp.ClientSession() as session:
                async with session.get(path) as response:
                    mime_type = response.headers.get("Content-Type")
                    mine_byte = await response.read()
                    if str(mine_byte).startswith("b'{\"code\":404"):
                        return 404

                    base64_encoded = base64.b64encode(mine_byte).decode("utf-8")
                    return f"data:{mime_type};base64,{base64_encoded}"

    @staticmethod
    def to_user_agent(ua: str) -> str:
        """
        浏览器UA简写。

        Args:
            ua (str): 字段名。

        Returns:
            str: 简写的UA名。

        Author:
            zero
        """
        ua_dict = {
            "chrome": "Chrome",
            "firefox": "Firefox",
            "safari": "Safari",
            "opera": "Opera",
            "edge": "Edge",
            "wechat": "MicroMessenger"
        }

        ua = ua.lower()
        for key, value in ua_dict.items():
            if value in ua:
                return key

        if "msie" in ua or "trident/" in ua:
            return "ie"

        return "other"

    @classmethod
    def generate_thumbnail(cls, input_path: str, output_path: str, width: int = None, height: int = None):
        """
        生成缩略图

        Args:
            input_path (str): 原始路径(/www/images/test.png)
            output_path (str): 输出路径(/server/public/storage/drawing/thumbnail/20250208/test.png)
            width (int): 裁剪后的图像宽度
            height (int): 裁剪后的图像高度

        Author:
            zero
        """
        try:
            if not os.path.isdir(os.path.dirname(output_path)):
                os.makedirs(os.path.dirname(output_path), exist_ok=True)

            with Image.open(input_path) as img:
                # 缩略图默认宽度
                thumbnail_width = 350 if width is None else width
                # 原始图像的宽高
                original_width, original_height = img.size
                # 计算缩略图高度
                thumbnail_height = height
                if height is None:
                    thumbnail_height = int(original_height * thumbnail_width / original_width)
                # 生成缩略图
                thumbnail_image = img.resize((thumbnail_width, thumbnail_height), Image.Resampling.LANCZOS)

                # 保存缩略图
                thumbnail_image.save(output_path, format="PNG")
        except Exception as e:
            raise Exception(f"生成缩略图时出错: {str(e)}")
